﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace MqttSample
{
    /// <summary>
    /// 数据通道
    /// </summary>
    public class DataChannelWrapper : IDisposable
    {
        private int _consumeThreadCount = 1;
        private List<Task> _taskList;
        private CancellationTokenSource _cts;
        private static readonly Lazy<DataChannelWrapper> _instance = new Lazy<DataChannelWrapper>(() => new DataChannelWrapper());
        private Channel<(MqttClientOption, CallbackData)> _dataChannel = Channel.CreateUnbounded<(MqttClientOption, CallbackData)>();

        public static DataChannelWrapper Instance = _instance.Value;
        public event Action<MqttClientOption, CallbackData> ReadEvent;
        
        private DataChannelWrapper()
        {
            _cts = new CancellationTokenSource();
            _taskList = new List<Task>();
        }

        /// <summary>
        /// 初始化消费线程
        /// </summary>
        /// <param name="consumeThreadCount">默认消费线程为1</param>
        public void Init(int consumeThreadCount = 1)
        {
            _consumeThreadCount = consumeThreadCount <= 1 ? 1 : consumeThreadCount;
            Disposing();
            for (int i = 0; i < _consumeThreadCount; i++)
            {
                var task = new Task(Do, _cts.Token, TaskCreationOptions.LongRunning);
                _taskList.Add(task);
                task.Start();
            }
        }

        /// <summary>
        /// 写数据
        /// </summary>
        public async Task WriteAsync(MqttClientOption option, CallbackData data)
        {
            await _dataChannel.Writer.WriteAsync((option, data), _cts.Token);
        }

        #region IDisposable
        public void Dispose()
        {
            _cts.Cancel();
            Disposing();
        }

        private void Disposing()
        {
            _taskList.ForEach((task) =>
            {
                task?.Dispose();
            });
            _taskList.Clear();
        }
        #endregion

        #region private methods
        /// <summary>
        /// 数据处理
        /// </summary>
        private async void Do()
        {
            while (!_cts.IsCancellationRequested)
            {
                var tupleData = await _dataChannel.Reader.ReadAsync();
                if (tupleData.Item1 != null && tupleData.Item2 != null)
                {
                    ReadEvent?.Invoke(tupleData.Item1, tupleData.Item2);
                }

            }
        }
        #endregion
    }
}
